package com.example.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.concurrent.Phaser;

@Service
@Slf4j
public class DataPipelineService {

    private final Phaser phaser = new Phaser(1); // 主线程注册到Phaser

    public void executeDataPipeline() {
        // 启动三个线程，每个线程执行读取、处理和写入数据的任务
        for (int i = 0; i < 3; i++) {
            phaser.register();
            new Thread(new DataPipelineWorker(i)).start();
        }

        // 主线程在每个阶段之间等待所有子线程完成
        phaser.arriveAndAwaitAdvance(); // 等待所有线程完成读取阶段
        log.info("所有线程完成数据读取");

        phaser.arriveAndAwaitAdvance(); // 等待所有线程完成处理阶段
        log.info("所有线程完成数据处理");

        phaser.arriveAndAwaitAdvance(); // 等待所有线程完成写入阶段
        log.info("所有线程完成数据写入");

        // 终止Phaser
        phaser.arriveAndDeregister();
        log.info("数据处理管道完成");
    }

    private class DataPipelineWorker implements Runnable {
        private final int workerId;

        public DataPipelineWorker(int workerId) {
            this.workerId = workerId;
        }

        @Override
        public void run() {
            try {
                // 读取数据阶段
                log.info("线程 {} 正在读取数据", workerId);
                Thread.sleep(1000 + workerId * 500L); // 模拟读取操作
                phaser.arriveAndAwaitAdvance(); // 等待所有线程完成读取

                // 处理数据阶段
                log.info("线程 {} 正在处理数据", workerId);
                Thread.sleep(1000 + workerId * 500L); // 模拟处理操作
                phaser.arriveAndAwaitAdvance(); // 等待所有线程完成处理

                // 写入数据阶段
                log.info("线程 {} 正在写入数据", workerId);
                Thread.sleep(1000 + workerId * 500L); // 模拟写入操作
                phaser.arriveAndAwaitAdvance(); // 等待所有线程完成写入

                phaser.arriveAndDeregister(); // 任务完成，注销自己
                log.info("线程 {} 已注销", workerId);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}
